 
/* 
 * Copyright (C) Ignacio Lucas
 */
 
/*
 *  Librería que encpasula el manejo de sockets no bloqueantes con el cambio de contexto
 *  a otra corrutina cuando se detecta que una llamada a alguna operacion de socket (connect, read, write)
 *  bloquearia al hilo de ejecucion.
 *
 *  Ignacio Lucas <ignacio.lucas@gmail.com>
 *
 */
#include "../../common.h"
//#include "../net.h"
//#include "../../server_worker.h"
#include "../../socket.h"
#include "../../job_event.h"
#include "../../worker.h"
#include <fcntl.h>
#include <sys/ioctl.h>

#define ADDRINFO_LENGTH sizeof(struct addrinfo)

// estructura asoiada al socket.
typedef struct s_socket
{
	int fd;
	int opened;
	int peer_closed;
	int last_error;
	int server;
	int domain;
	int type;
	int protocol;
	server_worker_job_t jobt;

	//enough_read_bytes read_callback;
	int registered_events;

	struct sockaddr_in address; // asumo IPv4.

	// configuraciones de timeout de sockets.
	timer_delta_t connect_timeout;
	timer_delta_t read_timeout;
	timer_delta_t write_timeout;
	
}	socket_p;


/*
	http://www.unix.com/high-level-programming/36919-read-write-socket-error.html
	
	Sockets and errors
	
	
	Here are some examples of when these errors occur:
	write(..) on a socket that has been closed at the other end will cause a SIGPIPE.
	
	read on a remotely closed socket will return 0 (EOF) or ECONNRESET in some cases, see the last part of this reply.
	
	read(..) or write(..) on a socket that was never connected will return ENOTCONN.
	
	read(..) or write(..) on a locally closed socket will return EBADF
	
	I'm referring to [comp.unix.programmer] Unix-socket-faq for network programming when answering the question regarding read returning 0 or ECONNRESET:
	
	"If the peer calls close() or exits, without having messed with
	SO_LINGER, then our calls to read() should return 0. It is less clear
	what happens to write() calls in this case; I would expect EPIPE, not
	on the next call, but the one after.
	
	If the peer reboots, or sets l_onoff = 1, l_linger = 0 and then
	closes, then we should get ECONNRESET (eventually) from read(), or
	EPIPE from write()."
*/

/*
static int socket_make_blocking(socket_p *s)
{
	int flags=0;
	
	if (((flags = fcntl(s->fd, F_GETFL, 0)) < 0 ||
		 fcntl(s->fd, F_SETFL, flags & ~O_NONBLOCK) < 0)) {
		s->last_error=errno;
		return 1;
	}
	return 0;
}
*/

size_t socket_get_struct_size()
{
	return sizeof(socket_p);
}

int socket_set_registered_events(socket_t st, int rev)
{
	((socket_p *)st)->registered_events = rev;
	return 0;
}

int socket_get_registered_events(socket_t st)
{
	return ((socket_p *)st)->registered_events;
}

// configuracion de un socket en modo no bloqueante.
static int socket_make_non_blocking(socket_p *s)
{
	int flags=0;
	/*
		Sockets can be set to either blocking or nonblocking I/O mode. 
		The FIONBIO ioctl operation is used to determine this mode. 
		When the FIONBIO ioctl is set, the socket is marked nonblocking. 
		If a read is tried and the desired data is not available, the socket does not wait for the data to become available, 
		but returns immediately with the EWOULDBLOCK error code.
		Note: 		The EWOULDBLOCK error code is defined with the _BSD define and is equivalent to the EAGAIN error code.
		
		When the FIONBIO ioctl is not set, the socket is in blocking mode. 
		In this mode, if a read is tried and the desired data is not available, 
		the calling process waits for the data. 
		Similarly, when writing, if FIONBIO is set and the output queue is full, 
		an attempt to write causes the process to return immediately with an error code of EWOULDBLOCK.
		When performing nonblocking I/O on sockets, a program must check for the EWOULDBLOCK error code 
		(stored in the errno global variable). This occurs when an operation would normally block, 
		but the socket it was performed on is marked as nonblocking. 
		The following socket subroutines return a EWOULDBLOCK error code:
	    * accept
	    * send
	    * recv
	    * read
	    * write
	
		Processes using these subroutines should be prepared to deal with the EWOULDBLOCK error code. 
		For a nonblocking socket, the connect subroutine returns an EINPROGRESS error code.
	*/
	
	/*ioctl(s->fd, FIONBIO, &flags) && */
	if (((flags = fcntl(s->fd, F_GETFL, 0)) < 0 ||
		 fcntl(s->fd, F_SETFL, flags | O_NONBLOCK) < 0)) {
		s->last_error=errno;
		return 1;
	}
	return 0;
}

// creacion de un socket.
static socket_t socket_create(socket_t st, int domain, int type, int protocol, server_worker_job_t jobt)
{
	int sfd;
	socket_p *s = (socket_p *)st;


	// fd>=0 se toma como valor valido.
	s->fd=-1;
	
	sfd = socket(domain, type, protocol);

	if (sfd == -1)
	{
		logger_core_error("[socket] error al crear socket. Domain: %d, type: %d, protocol: %d, error: %s", domain, type, protocol, strerror(errno));
		return NULL;
	}

	//flags = 1;

	s->opened=0;
	s->peer_closed=0;
	s->fd=sfd;
	s->jobt=jobt;
	s->domain=domain;
	s->type=type;
	s->protocol=protocol;
	
	//s->read_callback = NULL;
	s->registered_events = 0;
	s->connect_timeout = SOCKET_INFINITE_TIMEOUT; // inicialmente timeout infinito.
	s->read_timeout = SOCKET_INFINITE_TIMEOUT; // inicialmente timeout infinito.
	s->write_timeout = SOCKET_INFINITE_TIMEOUT; // inicialmente timeout infinito.

	memset(&(s->address), 0, sizeof(s->address));
	
	if (socket_make_non_blocking(s) != 0)
	{
		// solo cerramos el socket. La liberación de memoria se hace fuera.
		socket_close(s);
		return NULL;
	}
	return (socket_t)s;
}

socket_t socket_init_with_job(int domain, int type, int protocol, server_worker_job_t jobt)
{
	socket_p *s;
	socket_t ret;
	if ((s = ((socket_p *)common_objects_pool_get(SOCKET_POOL))) == NULL)
	{
		return NULL;
	}
	ret=socket_create(s,domain, type, protocol, jobt);
	if (ret == NULL)
	{
		common_objects_pool_put(SOCKET_POOL, s);
	}
	return ret;
}

socket_t socket_init(int domain, int type, int protocol)
{
	return socket_init_with_job(domain, type, protocol, NULL);
}


int socket_set_connect_timeout(socket_t st, timer_delta_t timeout)
{
	socket_p *s = (socket_p *)st;
	s->connect_timeout = timeout;
	return 0;
}

timer_delta_t socket_get_connect_timeout(socket_t st)
{
	return ((socket_p *)st)->connect_timeout;
}

int socket_set_read_timeout(socket_t st, timer_delta_t timeout)
{
	socket_p *s = (socket_p *)st;
	s->read_timeout = timeout;
	return 0;
}

timer_delta_t socket_get_read_timeout(socket_t st)
{
	return ((socket_p *)st)->read_timeout;
}

int socket_set_write_timeout(socket_t st, timer_delta_t timeout)
{
	socket_p *s = (socket_p *)st;
	s->write_timeout = timeout;
	return 0;
}

timer_delta_t socket_get_write_timeout(socket_t st)
{
	return ((socket_p *)st)->write_timeout;
}

int socket_set_server_worker_job(socket_t st, server_worker_job_t jobt)
{
	((socket_p *)st)->jobt=jobt;
	return 0;
}

server_worker_job_t socket_get_server_worker_job(socket_t st)
{
	return ((socket_p *)st)->jobt;
}

/*enough_read_bytes socket_get_read_callback(socket_t st)
{
	return ((socket_p *)st)->read_callback;
}*/

/*int socket_set_read_callback(socket_t st, enough_read_bytes read_callback)
{
	((socket_p *)st)->read_callback = read_callback;
	return 0;
}*/


socket_t socket_reuse_instance(socket_t st, int domain, int type, int protocol, server_worker_job_t jobt)
{
		if (socket_is_open(st))
		{
			socket_close(st);
		}
		return socket_create(st, domain, type, protocol, jobt);
}

int socket_is_open(socket_t st)
{
	return ((socket_p *)st)->fd > -1;
}

int socket_close(socket_t st)
{
	return close(((socket_p *)st)->fd);
}

int socket_destroy(socket_t st)
{
	if (st == NULL)
	{
		return 1;
	}
	if (socket_is_open(st))
	{
		socket_close(st);
	}
	
	((socket_p *)st)->fd = -1;
	((socket_p *)st)->jobt = NULL;
	common_objects_pool_put(SOCKET_POOL, st);
	return 0;
}

int socket_connect(socket_t st, char *server, u_short port)
{
	socket_p *s = (socket_p *)st;
	char str_port[15];

	int ret = 0, reuseaddr = 1;
	int find_dir=0;
	job_event_t event;
   	struct addrinfo *result;
	struct addrinfo hint;

	socket_config sock_config; // configuraciones propias para socket.
	int events_config;




	// initializacion de hint para getaddrinfo.
	memset(&(hint), 0, ADDRINFO_LENGTH);
	hint.ai_family = s->domain;
	hint.ai_socktype = s->type;
	hint.ai_protocol = s->protocol;
	hint.ai_flags = 0;

	sprintf(str_port, "%d", port);

	find_dir=getaddrinfo(server, str_port, &(hint), &result);

	if (find_dir != 0)
	{
		// no se encontró la dirección. salimos con error.
		logger_core_info("[socket] server:port not found: %p:%d", server, port);
		s->last_error=1;
		socket_close(s);
		return 1;
	}
	// resguardamos la primer dirección encontrada y liberamos memoria. Asumimos IPv4. Si se quiere hacer la version general hay que usar result->ai_addrlen como ultimo parametro del memcpy, usando en s->address un sockaddr * generico.
	memcpy(&(s->address), result->ai_addr, sizeof(s->address));

	// ya podemos liberar la memoria del getaddrinfo
	freeaddrinfo(result);
	
	// Asignamos el flag de reuse address
	if (setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr, sizeof(reuseaddr)) < 0)
	{
		s->last_error=errno;
		return 2;
	}


	//if (connect(s->fd, s->address.ai_addr, s->address.ai_addrlen) == -1) {
	if (connect(s->fd, (struct sockaddr *)&(s->address), sizeof(s->address)) == -1) {
		s->last_error=errno;
		
		if (s->last_error != EWOULDBLOCK && s->last_error != EINPROGRESS)
		{
			socket_close(s);
			return 3;
		}
			
		// Preparo para detener el job hasta que se complete el connect.	
		if (s->jobt != NULL)
		{
			// configuro interes en el evento de OUTPUT (y opcional el timeout).
			events_config = JOB_EVENT_OUTPUT;
			sock_config.st = s;
			sock_config.timeout_config.delta_timeout = socket_get_connect_timeout(s);
			if (sock_config.timeout_config.delta_timeout != SOCKET_INFINITE_TIMEOUT)
			{
				events_config |= JOB_EVENT_TIMEOUT;
			}
			
			ret=server_worker_job_register_event_and_stop(s->jobt, JOB_SOURCE_SOCKET, (void *)&(sock_config), events_config);
			if (ret != 0)
			{
				logger_core_error("[socket] register_event_and_stop %d", ret);
				socket_close(s);
				return ret;
			}
		
			// Chequeo resultados al volver a retomar la tarea.
			event = server_worker_job_get_outstanding_event(st->jobt);
			if (event->fired_events & (JOB_EVENT_ERR | JOB_EVENT_HUP))
			{
				s->last_error=event->fired_events;
				logger_core_error("[socket] fired_events con error o hup %d", event->fired_events);
				socket_close(s);
				return 4;
			}

			// Verificamos si salimos por timeout.
			if (server_worker_job_get_outstanding_event(st->jobt)->fired_events & (JOB_EVENT_TIMEOUT))
			{
				logger_core_info("[socket_read] timeout de connect. FD: %d",st->fd);
				return 5;
			}

			// Registro socket abierto.
			s->opened=1;
			s->server=0;
			s->peer_closed=0;
		}
		else
		{
			socket_close(s);
			return 5;
		}
	}
	return 0;
}

int socket_get_fd(socket_t st)
{
	return ((socket_p *)st)->fd;
}


/**
	La semántica de socket_read es:
	
	Se lee hasta completar maxlong. Si antes de llegar a maxlong se detecta que se va a bloquear se procede como sigue:
	
	- Si el total leido hasta el momento es cero, se detiene el job.
	- Si el total leido hasta el momento es mayor a cero, se devuelve lo que se leyo, siendo responsabilidad de la aplicaci�n
	  de verificar lo recibido e invocar nuevamente a socket_read de ser necesario.
**/
	
int socket_read(socket_t st, char *ptr, int maxlong)
{
	int	n=0, leido=0, ret = 0;
	socket_p *s = (socket_p *)st;

	socket_config sock_config; // configuraciones propias para socket.
	int events_config;

	
	while (n < maxlong){
	//while ((n < maxlong) && (n == 0)){
	//while (n == 0){
		//errno=0;
		leido = read(s->fd, ptr + n, maxlong-n);
		
		//logger_core_info("[socket] LEIDO:%d, FD:%d", leido, socket_get_fd(st));
		
		if (leido == -1)
		{
			s->last_error=errno;
			// como si fuera leido == 0.
			if (s->last_error == ECONNRESET)
			{
				s->peer_closed=1;
				return n;
			}
			
			if ((s->last_error != 0) && (s->last_error != EWOULDBLOCK) && (s->last_error != EAGAIN)  && (s->last_error != EINPROGRESS))
			//if ((s->last_error != EWOULDBLOCK) && (s->last_error != EAGAIN))
			{
				logger_core_info("[socket] READ_ERROR: %d", s->last_error);
				return(-1);
			}
			else
			{
				// Preparo para detener la tarea hasta que haya bytes para leer.
				//if ((s->jobt != NULL) && ((s->read_callback == NULL) || ((s->read_callback != NULL) && (s->read_callback(ptr, n, maxlong) == 0))))
				
				// si no leimos ningun byte y hay contexto de job, detenemos la tarea.
				if ((s->jobt != NULL) && (n == 0))
				{
					// configuro interes en el evento de INPUT (y opcional el timeout).
					sock_config.st = st;
					events_config = JOB_EVENT_INPUT;
					sock_config.timeout_config.delta_timeout = socket_get_read_timeout(s);
					if (sock_config.timeout_config.delta_timeout != SOCKET_INFINITE_TIMEOUT)
					{
						events_config |= JOB_EVENT_TIMEOUT;
					}

					ret=server_worker_job_register_event_and_stop(st->jobt, JOB_SOURCE_SOCKET, (void *)&(sock_config), events_config);
					if (ret != 0)
					{
						logger_core_info("[socket_read] register_error. %d", ret);
						return -2;
					}
				
					// Chequeo resultados al retomar la tarea.
					if (server_worker_job_get_outstanding_event(st->jobt)->fired_events & (JOB_EVENT_ERR | JOB_EVENT_HUP))
					{
						logger_core_info("[socket_read] fired_errors. %d", server_worker_job_get_outstanding_event(st->jobt)->fired_events);
						return -3;
					}

					// Verificamos si salimos por timeout.
					if (server_worker_job_get_outstanding_event(st->jobt)->fired_events & (JOB_EVENT_TIMEOUT))
					{
						logger_core_info("[socket_read] timeout de read. FD: %d", st->fd);
						return n;
					}
				}
				else
				{
					// devuelvo bytes leidos.
					return n;
				}
			}
		}
		else
		{
			/* el otro par cerro la conexion. */
			if (leido == 0)
			{
				s->peer_closed=1;
				return n;
			}
			else
			{
				n+=leido;
			}
		}
	}
	return(n);
}

int socket_peer_closed(socket_t st)
{
	return ((socket_p *)st)->peer_closed;
}

int socket_write(socket_t st, char *ptr, int nbytes)
{
	int	nfaltan, nenviados, ret = 0;
	socket_p *s=(socket_p *)st;
	socket_config sock_config; // configuraciones propias para socket.
	int events_config;


	nfaltan = nbytes;
	while (nfaltan > 0)
	{
		nenviados = write(s->fd, ptr, nfaltan);
		if (nenviados == -1)
		{
			st->last_error=errno;
			if (st->last_error == EPIPE)
			{
				// el otro par cerro la conexion
				s->peer_closed=1;
				return -1;
			}
			if ((s->last_error != EWOULDBLOCK) && (s->last_error != EAGAIN) && (s->last_error != EINPROGRESS))
			{
				return(-2);
			}
			else
			{
				// Preparo a detener la tarea hasta que haya posibilidad de escribir.
				if (s->jobt != NULL)
				{
					// configuro interes en el evento de OUTPUT (y opcional el timeout).
					events_config = JOB_EVENT_OUTPUT;
					sock_config.st = s;
					sock_config.timeout_config.delta_timeout = socket_get_write_timeout(s);
					if (sock_config.timeout_config.delta_timeout != SOCKET_INFINITE_TIMEOUT)
					{
						events_config |= JOB_EVENT_TIMEOUT;
					}

					ret=server_worker_job_register_event_and_stop(s->jobt, JOB_SOURCE_SOCKET, (void *)&(sock_config), events_config);
					if (ret != 0)
					{
						logger_core_error("[socket_write] error registrando eventos. %d", ret);
						return -3;
					}
					
				
					// Chequeo resultados luego de retomar la tarea.
					if (server_worker_job_get_outstanding_event(s->jobt)->fired_events & (JOB_EVENT_ERR | JOB_EVENT_HUP))
					{
						logger_core_error("[socket_write] error events: %d", server_worker_job_get_outstanding_event(s->jobt)->fired_events);

						return -4;
					}

					// Verificamos si salimos por timeout.
					if (server_worker_job_get_outstanding_event(st->jobt)->fired_events & (JOB_EVENT_TIMEOUT))
					{
						logger_core_info("[socket_read] timeout de write. FD: %d", st->fd);
						return -5;
					}

				}
				else
				{
					return -5;
				}
			}
		}
		else
		{
			nfaltan -= nenviados;
			ptr   += nenviados;
		}
	}
	return(nbytes - nfaltan);
}

int socket_listen(socket_t st, char *listen_ip, int listen_port, int queue_size)
{
	int ret = 0, reuseaddr = 1;
	socket_p *s = (socket_p *)st;
	int find_dir=0;
	char str_listen_port[15];
	struct addrinfo *result;
	struct addrinfo hint;


	// initializacion de hint usado por getaddrinfo.
	memset(&(hint), 0, ADDRINFO_LENGTH);
	hint.ai_family = s->domain;
	hint.ai_socktype = s->type;
	hint.ai_protocol = s->protocol;
	hint.ai_flags = 0;


	/*if (!socket_is_open(st))
	{
		return 1;
	}*/
 	

	/*			INICIALIZACION DEL SERVER
	
	* 		Inicializar la estructura del socket:
	* bzero(): borrar la estructura
	* htonl(): Host-TO-Network Long integer (Byte order)
	* htons(): Host-TO-Network Short integer (Byte order)     
	*/

	sprintf(str_listen_port, "%d", listen_port);

	find_dir=getaddrinfo(listen_ip, str_listen_port, &(hint), &result);
	if (find_dir != 0) {
		logger_core_info("[socket] server ip:port (%s:%s) not found.", listen_ip, str_listen_port);
		s->last_error=errno;
		return 1;
	}

	//resguardamos la primer dirección y liberamos memoria. Asumimos IPv4. Si se quiere hacer la version general hay que usar result->ai_addrlen como ultimo parametro del memcpy, usando en s->address un sockaddr * generico.
	memcpy(&(s->address), (result->ai_addr), sizeof(s->address));
	freeaddrinfo(result);

	if (setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr, sizeof(reuseaddr)) < 0)
	{
		s->last_error=errno;
		return 1;
	}
	/*
	 * 		Vincular el socket con la direccion local 
	 */
	//if ((ret=(bind(s->fd, s->address.ai_addr, s->address.ai_addrlen))) < 0)
	if ((ret=(bind(s->fd, (struct sockaddr *)&(s->address), sizeof(s->address)))) < 0)
	{
		s->last_error=errno;
		return (2); /* error: no se puede hacer el bind */
	}
	ret=listen(s->fd, queue_size);
	if (ret < 0)
	{
		s->last_error=errno;
		return 3;
	}
	
	if (socket_make_non_blocking(s) != 0)
	{
		return 4;
	}
	
	s->server=1;
	return 0;
} 

socket_t socket_accept(socket_t st)
{
	socket_p *s=(socket_p *)st, *client_socket = NULL;
	int client_fd=0;
	size_t clilen=0;
	struct sockaddr_in	cli_addr;
	
	if (!s->server)
	{
		return NULL;
	}
	client_fd = accept(s->fd, (struct sockaddr *) &cli_addr, &clilen);
	if (client_fd < 0)
	{
		s->last_error = errno;
		return NULL;
	}
	
	client_socket=(socket_p *)common_objects_pool_get(SOCKET_POOL);
	if (client_socket == NULL)
	{
		s->last_error = errno;
		close(client_fd);
		return NULL;
	}
	
	// Registramos sockaddr.
	//memcpy(&(client_socket->address.ai_addr), (struct sockaddr *) &cli_addr, clilen);
	//client_socket->address.ai_addrlen=clilen;

	client_socket->fd = client_fd;
	client_socket->opened=1;
	client_socket->peer_closed=0;
	client_socket->last_error=0;
	client_socket->server=0;
	client_socket->jobt=NULL;
	client_socket->registered_events = 0;
	client_socket->connect_timeout = SOCKET_INFINITE_TIMEOUT; // timeout infinito.
	client_socket->read_timeout = SOCKET_INFINITE_TIMEOUT; // timeout infinito.
	client_socket->write_timeout = SOCKET_INFINITE_TIMEOUT; // timeout infinito.

	//client_socket->read_callback = NULL;
	
	// Hacemos no bloqueante.
	if (socket_make_non_blocking(client_socket) != 0)
	{
		s->last_error = socket_get_last_error(client_socket);
		socket_destroy(client_socket);
		return NULL;
	}
	return (socket_t)client_socket;
}

int socket_get_last_error(socket_t st)
{
	return ((socket_p *)st)->last_error;
}
